package com.cong

import org.apache.spark.{SparkContext, SparkConf}

/**
 * Created by DELL on 2018/2/12.
 */
object ProcessPhone {

  def main(args: Array[String]) {

         val conf=  new SparkConf().setAppName("localphone1").setMaster("local[2]");
         val sc=  new SparkContext(conf)
        val source= sc.textFile("F:\\bs_log")
       //val tt=source.map(_.split(",")).map(x=>(x(0),x(1)))

      val rdd1= source.map(line=>{
         val mobile=line.split(",")
         val p=mobile(0)
         val z=mobile(2)
        val evnet=mobile(3)

         val t=if(evnet == "1") -mobile(1).toLong else  mobile(1).toLong

         val pp=p+"_"+z
         (pp,t)

       })

     val rdd2=rdd1.groupBy(_._1).mapValues(_.foldLeft(0L)(_+_._2))

    val rdd3=rdd2.map(x=>{
        val arr=x._1.split("_")
        val m=arr(0)
        val lac=arr(1)
        val time=x._2
      (m,lac,time)
    })

     val rdd4=rdd3.groupBy(_._1)
    val rdd5=rdd4.mapValues(t=>t.toList.sortBy(_._3).reverse.take(2))
     println(rdd5.collect().toBuffer)
    sc.stop()
  }

}
